package com.bd08.flink.demo.jing.table

import org.apache.flink.api.scala.{ExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.Expressions.{$, row}
import org.apache.flink.table.api.{DataTypes, Schema}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Demo2 extends App {

  // 创建批执行上下文
  //val benv = ExecutionEnvironment.getExecutionEnvironment

  // 创建 流执行上下文
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  // 创建 流表执行上下文
  val stenv = StreamTableEnvironment.create(env)
  // 变化 csv 中数据 为 row  注册成表
  val ds = env.readTextFile("E:\\flink_workspace\\tutuDai\\src\\main\\resources\\customers.csv")

  val ds2 = env.readTextFile("E:\\flink_workspace\\tutuDai\\src\\main\\resources\\orders.csv")
  val rsds2 = ds2.map(line => {
    val fields = line.split("[\",]{1,3}")
    val rs = fields.toList.tail
    (rs(0),rs(1),rs(2),rs(3))
  })

  val rsds = ds
    .map(line => {
      val fields = line.split("[\",]{1,3}")
      val rs = fields.toList.tail
      (rs(0),rs(1),rs(2),rs(3),rs(4),rs(5),rs(6),rs(7),rs(8))
    })
  //stenv.createTemporaryView("customer",rsds)
//  val schame = Schema.newBuilder()
//    .fromRowDataType(
//      DataTypes.ROW(
//        DataTypes.FIELD("name", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("age", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("birth", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("birth1", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("birth2", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("birth3", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("birth4", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("birth5", DataTypes.STRING().notNull())
//        ,
//        DataTypes.FIELD("birth6", DataTypes.STRING().notNull())
//    ))
//    .build()
  stenv.createTemporaryView("customer",rsds)
  stenv.createTemporaryView("order",rsds2)


  //stenv.executeSql("create table customerWithSchema(id string,name string,name1 string,name2 string,name3 string,name4 string,name5 string,name6 string,name7 string) with ( 'connector' = 'filesystem','path' = 'D:\\yahu\\bugeikan\\src\\main\\resources\\rscu','format' = 'file' )")
  // 起 列别名
  val ct = stenv.from("customer").as("name1","name2","name3","name4","name5","name6","name7","name8","name9")

  val ct2 = stenv.from("order").as("id","data","id2","status")
//  ct.select($("*")).limit(5).execute().print()
//
//  ct2.select($("*")).limit(5).execute().print()
  ct.join(ct2).limit(5).printSchema()
//  stenv.executeSql("select c.name1,c.name2,c.name3 from customer c join order o on c.name1=o.id limit 5").print()


//  ct.executeInsert("customerWithSchema");
//  stenv.executeSql("select * from customerWithSchema limit 5")


  // 利用table 或者 sql api 处理数据
  //env.execute()


}
